package com.xxx.rabbitmq.listener;

import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.amqp.rabbit.annotation.*;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;

@Slf4j
@Component
public class RabbitMqListener {

    /**
     * ①简单发送
     */
    @RabbitListener(queues = "simple.queue")
    public void listenSimpleRabbitMQ(String msg){
        log.info("接收到simple.queue队列消息：【"+msg+"】");
        // 模拟发生异常时，重新入队
//        throw new RuntimeException("发生异常，回执nack");
    }
//    @RabbitListener(queues = "simple.queue")
//    public void listenSimpleRabbitMQ(Message message, Channel channel){
//        String msg = new String(message.getBody());
//        log.info("接收到simple.queue队列消息：【"+msg+"】");
//
//        // 手动ack
//        long deliveryTag = message.getMessageProperties().getDeliveryTag();
//        try {
//            channel.basicAck(deliveryTag, false);
//            log.info("手动ack消息：【"+deliveryTag+"】");
//        } catch (IOException e) {
//            throw new RuntimeException(e);
//        }
//    }

    ///////////////////////////////// direct消息 start ///////////////////////////////
    /**
     * 在创建RabbitMQ Listener时创建队列并绑定到交换机
     * bindings = @QueueBinding():声明并绑定关系
     * value = @Queue():队列名称
     * exchange = @Exchange():交换机名称，发布订阅模式
     * key = {}:指定路由key
     */
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "direct.queue1"),
            exchange = @Exchange(name = "direct.exchange"),
            key = {"red", "blue"}))
    public void listenDirectQueue1(String msg) throws InterruptedException {
        log.info("接收到direct.queue1队列消息：【"+msg+"】");
        Thread.sleep(20);
    }

    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "direct.queue2"),
            exchange = @Exchange(name = "direct.exchange"),
            key = {"red", "yellow"}
    ))
    public void listenDirectQueue2(String msg) throws InterruptedException {
        log.info("接收到direct.queue2队列消息：【"+msg+"】");
        Thread.sleep(200);
    }
    ///////////////////////////////// direct消息 end ///////////////////////////////


    /////////////////////////////// fanout消息 start ///////////////////////////////
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "fanout.queue1"),
            exchange = @Exchange(name = "fanout.exchange", type = ExchangeTypes.FANOUT)
    ))
    public void listenFanoutQueue1(String msg) {
        log.info("接收到fanout.queue1队列消息：【"+msg+"】");
    }

    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "fanout.queue2"),
            exchange = @Exchange(name = "fanout.exchange", type = ExchangeTypes.FANOUT)
    ))
    public void listenFanoutQueue2(String msg) {
        log.info("接收到fanout.queue2队列消息：【"+msg+"】");
    }
    /////////////////////////////// fanout消息 end ///////////////////////////////


    ///////////////////////////////// topic消息 start ///////////////////////////////
    /**
     * TOPIC类型交换机
     * # 匹配1个或多个单词，如：china.# 匹配：china.spu.nb、china.item.spu
     * * 匹配1个单词，如：china.* 匹配：china.spu
     */
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "topic.queue1"),
            exchange = @Exchange(name = "topic.exchange", type = ExchangeTypes.TOPIC),
            key = {"china.#"}))
    public void listenTopicQueue1(String msg) {
        log.info("接收到topic.queue1队列消息：【"+msg+"】");
    }

    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "topic.queue2"),
            exchange = @Exchange(name = "topic.exchange", type = ExchangeTypes.TOPIC),
            key = {"#.news"}
    ))
    public void listenTopicQueue2(String msg) {
        log.info("接收到topic.queue2队列消息：【"+msg+"】");
    }
    ///////////////////////////////// topic消息 start ///////////////////////////////


    ///////////////////////////////// 过期消息 start ///////////////////////////////
    /**
     * 监听过期消息队列
     */
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "ttl.queue", durable = "true"),
            exchange = @Exchange(name = "ttl.exchange", delayed = "true"),
            key = "ttl"
    ))
    public void listenTtlQueue(String msg) {
        log.info("接收到过期消息队列ttl.queue队列消息：【"+msg+"】");
    }
    ///////////////////////////////// 过期消息 end ///////////////////////////////


    ///////////////////////////////// 延时队列 start ///////////////////////////////
    /**
     * 监听延时消息队列
     */
    @RabbitListener(queues="delay.queue")
    public void listenDelayQueue(String msg, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long tag) {
        log.info("接收到delay.queue延时队列消费 tag = {}, msg: {}", tag, msg);
    }
    ///////////////////////////////// 延时队列 end ///////////////////////////////

    ///////////////////////////////// 多个消费者同时消费同一个队列 start ///////////////////////////////
    /**
     * 监听消息队列：work.queue
     */
    @RabbitListener(queues="work.queue")
    public void listenWorkQueue1(String msg) throws InterruptedException {
        log.info("消费者1接收到work.queue队列消费: msg: {}", msg);
        Thread.sleep(20);
    }
    @RabbitListener(queues="work.queue")
    public void listenWorkQueue2(String msg) throws InterruptedException {
        log.info("消费者2接收到work.queue队列消费: msg: {}", msg);
        Thread.sleep(200);
    }
    ///////////////////////////////// 多个消费者同时消费同一个队列 end ///////////////////////////////


    ///////////////////////////////// lazy queue start ///////////////////////////////
    /**
     * 监听lazy queue队列
     */
    @RabbitListener(queuesToDeclare = @Queue(
            name = "lazy.queue",
            durable = "true",
            arguments = @Argument(name="x-queue-mode", value="lazy")
    ))
    public void listenLazyQueue(String msg) {
        log.info("接收到Lazy消息队列lazy.queue消息：【"+msg+"】");
    }
    ///////////////////////////////// lazy queue end ///////////////////////////////


}
